package com.ebspp.client.serverManager;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;

import com.ebspp.global.EBSPPTimer;
import com.ebspp.global.General;
import com.ebspp.thrift.Identity;
import com.ebspp.thrift.MsgTransfer;
import com.ebspp.thrift.ResourceType;

/**
 * MsgConnection is the heartbeat connection. 
 * It exchange information with server, and figure out the server's available resources;
 * It also monitors the hearbeat response time, which can be used to guess the server's physical location
 * 
 * @author x23pan
 *
 */
public class MsgConnection
{
	private boolean stop = false;
	
	private String serverIP;
	private MsgTransfer.Client msgClient;
	private TTransport msgTransport;

	private IStat stat;
	Map<ResourceType, Integer> resMap;
	Object mutex = new Object();
	
	Thread heartBeatT = null;
	
	public MsgConnection(String serverIP, int mPort, IStat stat) throws TTransportException
	{
		this.serverIP = serverIP;
		
		this.stat = stat;
		resMap = Collections.synchronizedMap(new HashMap<ResourceType, Integer>());
		
		TSocket socket = new TSocket(serverIP, mPort);
		msgTransport = new TFramedTransport(socket);
		TProtocol protocol = new TBinaryProtocol(msgTransport);
		msgClient = new MsgTransfer.Client(protocol);
		msgTransport.open();
	}
	
	public void initHeartBeat()
	{
		heartBeatT = new Thread(new HeartBeat());
		heartBeatT.start();
	}
	public boolean watch(ResourceType resType, Identity wID) throws TException
	{
		synchronized(mutex)
		{
			return msgClient.watch(resType, wID, General.getServerName());
		}	
	}
	
	public boolean deWatch(ResourceType resType, Identity wID) throws TException
	{
		synchronized(mutex)
		{
			return msgClient.deWatch(resType, wID, General.getServerName());
		}
	}
	
	
	public void close()
	{
		stop = true;
	}

	
	
	public boolean equals(Object obj)
	{
		if(obj == null)
			return false;
		if(obj instanceof MsgConnection)
		{
			MsgConnection conn = (MsgConnection)obj;
			if(serverIP.equals(conn.serverIP))
				return true;
		}
		return false;
	}
	
	public int hashCode()
	{
		return serverIP.hashCode() * 17;
	}
	
	class HeartBeat implements Runnable
	{

		@Override
		public void run() {
			// TODO Auto-generated method stub
			while(true)
			{
				try 
				{
					Map<ResourceType, Integer> availRes = null;
					synchronized(mutex)
					{
						long start = General.getT();
						availRes = msgClient.quest();
						stat.setMsgDelay(serverIP, General.getT() - start);
					}

					for(ResourceType type : availRes.keySet())			
						resMap.put(type, availRes.get(type));

					Thread.sleep(EBSPPTimer.heartBeatT);
				} 
				catch (TException e) 
				{
					// TODO Auto-generated catch block
					stat.setFatal(serverIP);
					e.printStackTrace();
					stop = true;
				} 
				catch (InterruptedException e) {
					// TODO Auto-generated catch block
					System.err.println("Unexpected interruption for message heart beat");
					e.printStackTrace();
					stop = true;
				}
				
				if(stop)
				{
					System.out.println("Message connections stopping: " + serverIP);
					msgTransport.close();
					return;
				}
			}
		}
		
	}
}
